package com.xq.tmall.agu.job;

import com.xq.tmall.agu.ferign.AbstractHtmlSpider;
import com.xq.tmall.agu.ferign.SinaIndexSpider;
import com.xq.tmall.agu.ferign.TencentIndexSpider;
import com.xq.tmall.agu.thread.MyThreadFactory;
import com.xq.tmall.entity.Stock;
import com.xq.tmall.service.StockService;
import com.xq.tmall.util.ITimeJobFactory;
import com.xq.tmall.util.SpringContextUtil;
import lombok.extern.slf4j.Slf4j;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author chenguoxiang
 * @create 2019-03-07 11:18
 **/
@Slf4j
public class IndexCrawlJob implements Runnable {

    private ScheduledExecutorService crawlExecutor;
    private  ScheduledExecutorService dbExecutor;
    private  BlockingQueue<Stock> stockQueue = new LinkedBlockingQueue<>(5000); // 使用有界队列防止内存溢出
    // 重试配置
    private static final int MAX_RETRIES = 3;
    private static final long RETRY_DELAY = 8000; // 5秒重试间隔
    // 任务跟踪
    private final Map<String, Future<?>> runningTasks = new ConcurrentHashMap<>();
    private final AtomicInteger activeTasks = new AtomicInteger(0);

    public IndexCrawlJob() {
        // 根据2核CPU优化线程池大小
        crawlExecutor = Executors.newScheduledThreadPool(4, new MyThreadFactory("crawler-index"));
        dbExecutor = Executors.newScheduledThreadPool(2, new MyThreadFactory("db-index"));
    }
    public void stopTask() {
        // 优雅关闭
        log.info("Shutting down crawler...");

        // 先停止接受新任务
        crawlExecutor.shutdown();
        dbExecutor.shutdown();

        try {
            // 等待现有任务完成
            if (!crawlExecutor.awaitTermination(30, TimeUnit.SECONDS)) {
                crawlExecutor.shutdownNow();
            }

            if (!dbExecutor.awaitTermination(30, TimeUnit.SECONDS)) {
                dbExecutor.shutdownNow();
            }

            // 等待队列处理完成
            while (!stockQueue.isEmpty()) {
                processStockBatch();
                Thread.sleep(1000);
            }

            log.info("All tasks completed");
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            log.error("Shutdown interrupted", e);
        }
    }

    private void processStocksInBatches(List<Stock> stocks, String market, boolean isSina) {
        List<String> batch = new ArrayList<>(300);
        for (int i = 0; i < stocks.size(); i++) {
            batch.add(stocks.get(i).getCode());

            if (i > 0 && (i % 300 == 0 || i == stocks.size() - 1)) {
                String taskId = market + "-" + (i / 300);
                scheduleCrawlTask(batch.toArray(new String[0]), isSina, taskId);
                batch.clear();

                try {
                    // 控制请求速率
                    Thread.sleep(8000);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }
        }
    }

    private void scheduleCrawlTask(String[] codes, boolean isSina, String taskId) {
        Runnable task = () -> {
            activeTasks.incrementAndGet();
            try {
                AbstractHtmlSpider<Stock> spider = isSina ?
                        new SinaIndexSpider(codes, stockQueue) :
                        new TencentIndexSpider(codes, stockQueue);

                // 带重试的执行
                executeWithRetry(spider);

            } finally {
                activeTasks.decrementAndGet();
                runningTasks.remove(taskId);
            }
        };

        Future<?> future = crawlExecutor.schedule(task, 0, TimeUnit.SECONDS);
        runningTasks.put(taskId, future);
    }

    private void executeWithRetry(Runnable task) {
        int retryCount = 0;
        while (retryCount <= MAX_RETRIES) {
            try {
                task.run();
                return; // 成功则退出
            } catch (Exception e) {
                retryCount++;
                if (retryCount <= MAX_RETRIES) {
                    log.warn("任务重试 (attempt {}/{}), retrying...", retryCount, MAX_RETRIES, e);
                    try {
                        Thread.sleep(RETRY_DELAY);
                    } catch (InterruptedException ie) {
                        Thread.currentThread().interrupt();
                        throw new RuntimeException("Retry interrupted", ie);
                    }
                } else {
                    log.error("最终重试后 {} attempts", MAX_RETRIES, e);
                }
            }
        }
    }

    private void processStockBatch() {
        try {
            int batchSize = Math.min(stockQueue.size(), 500);
            if (batchSize == 0) return;

            List<Stock> stockList = new ArrayList<>(batchSize);
            for (int i = 0; i < batchSize; i++) {
                Stock stock = stockQueue.poll();
                if (stock != null) {
                    Stock updateStock = new Stock();
                    updateStock.setCode(stock.getCode());
                    updateStock.setPrice(stock.getPrice());
                    updateStock.setYesterdayPrice(stock.getYesterdayPrice());
                    updateStock.setFluctuate( stock.getFluctuate());
                    updateStock.setPriceDate(stock.getPriceDate());
                    updateStock.setIndustry(stock.getIndustry());
                    updateStock.setTotalValue(stock.getTotalValue());
                    updateStock.setPb(stock.getPb());
                    updateStock.setPes(stock.getPes());
                    updateStock.setDividend(stock.getDividend());
                    updateStock.setAmplitude(stock.getAmplitude());
                    updateStock.setBiddingVolume(stock.getBvps());
                    stockList.add(updateStock);
                }
            }

            if (!stockList.isEmpty()) {
                StockService stockService = SpringContextUtil.getBean(StockService.class);
                executeWithRetry(() -> stockService.batchUpdateVolume(stockList));

            }
        } catch (Exception e) {
            log.error("Failed to process stock batch", e);
        }
    }

    @Override
    public void run() {

        try {
            // 确保之前的任务完成
            if (activeTasks.get() > 0) {
                log.warn("Previous tasks still running ({} active), skipping this run", activeTasks.get());
                return;
            }

            StockService stockService = SpringContextUtil.getBean(StockService.class);
            int count = stockService.getTodayCount();
            if (count > 2000) {
                log.info("今日股票已更新!");
                return;
            }
            // 分批处理SZ股票
            processStocksInBatches(stockService.getSZStockCodes(), "SZ", true);

            // 分批处理SH股票
            processStocksInBatches(stockService.getSHStockCodes(), "SH", false);

            // 分批处理CY股票
            processStocksInBatches(stockService.getCYStockCodes(), "CY", false);

            ITimeJobFactory.getJob(ITimeJobFactory.TIMEJOB.INDEX).execute(ITimerJob.COMMAND.STOP);

        } catch (Exception ex) {
            log.error("Crawl job failed", ex);
        }
    }
}
